package org.zjt.flink.stream;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.delta.DeltaFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.DeltaTrigger;
import org.apache.flink.table.shaded.org.apache.commons.codec.language.bm.Rule;
import org.junit.Test;

import javax.annotation.Nullable;
import java.util.Random;
import java.util.concurrent.TimeUnit;

import static junit.framework.TestCase.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;

/**
 * Description:
 *
 *      时间自带time。利用事件自己的time来数据处理。针对历史回放处理。
 *
 * @author juntao.zhang
 * Date: 2018-10-12 上午11:34
 * @see
 */
@Slf4j
public class WindowsTimeJob {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // EventTime 基于时间自身的时间，多用于回放历史数据。
        // ProcessTime 基于当前数据流收到的时间实时处理。

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        Random random = new Random();
        SingleOutputStreamOperator<Tuple2<Integer,Long>> reduce = env.addSource(new SourceFunction<Tuple2<Integer,Long>>() {
            @Override
            public void run(SourceContext<Tuple2<Integer,Long>> ctx) throws Exception {
                for (int i = 0; i < 1000; i++) {
                    Thread.sleep(100);
                    Tuple2<Integer, Long> integerLongTuple2 = new Tuple2<>();
                    integerLongTuple2.f0 = random.nextInt(10000);
                    integerLongTuple2.f1 = System.currentTimeMillis() - integerLongTuple2.f0 ;

                    ctx.collect(integerLongTuple2);

                }
            }

            @Override
            public void cancel() {

            }
        });

        TypeSerializer<Tuple2<Integer, Long>> serializer = reduce.getType().createSerializer(env.getConfig());
        reduce.assignTimestampsAndWatermarks(new CarTimestamp())
                .keyBy(1)

                .window(GlobalWindows.create())
                // 计算时间段
                .evictor(TimeEvictor.of(Time.of(10, TimeUnit.SECONDS)))
                .trigger(DeltaTrigger.of(5, new DeltaFunction<Tuple2<Integer, Long>>() {
                    @Override
                    public double getDelta(Tuple2<Integer, Long> oldDataPoint, Tuple2<Integer, Long> newDataPoint) {
                        return Math.abs(oldDataPoint.f0 - newDataPoint.f0);
                    }
                },serializer)).maxBy(0).print();


        env.execute();
    }



    private static class CarTimestamp implements AssignerWithPeriodicWatermarks<Tuple2<Integer, Long>> {


        private static Long maxOutOfOrderness = 10000L ; //最大允许的乱序时间是10s

        private long currentTimestamp = Long.MIN_VALUE;   //当前处理到的时间
        /**
         * 水印：
         *  1、保证刺顺序执行
         *  2、保证乱序等待最长时间
         * @return
         */
        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            return new Watermark(currentTimestamp - maxOutOfOrderness);
        }

        @Override
        public long extractTimestamp(Tuple2<Integer, Long> element, long previousElementTimestamp) {

            currentTimestamp = Long.max(currentTimestamp,element.f1);

            return currentTimestamp;
        }
    }





}
